package spike.pool.impl;

import spike.pool.ConnectionPool;
import spike.pool.ConnectionSource;

import java.util.ArrayDeque;
import java.util.Deque;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;

public class ConnectionPoolService<T> implements ConnectionPool<T> {

    private final Deque<T> freeConnectionsQueue = new ArrayDeque<>();
    private final Map<T, Long> timeToKillConnections = new HashMap<>();

    private final ConnectionSource<T> connectionSource;

    private final int minCapacity;
    private final int maxCapacity;
    private int aliveConnections;

    private boolean isServiceStarted = false;

    private final long keepAliveTime;

    private ConnectionKiller connectionKiller;

    public ConnectionPoolService(int min, int max, long keepAliveTime, TimeUnit unit, ConnectionSource<T> src) {
        this.keepAliveTime = unit.toNanos(keepAliveTime);

        this.connectionSource = src;
        this.aliveConnections = min;
        this.minCapacity = min;
        this.maxCapacity = max;

        for(int i = 0; i < minCapacity; i++) {
            freeConnectionsQueue.add(openNewConnection());
        }
    }

    public ConnectionPoolService(int capacity, ConnectionSource<T> src) {
        this(capacity, 0, 0, TimeUnit.SECONDS, src);
    }

    private synchronized T openNewConnection() {
        T connection = connectionSource.open();
        timeToKillConnections.put(connection, (long) 0);
        return connection;
    }

    private synchronized T mainGet(long timeout) throws InterruptedException {
        T connection;
        if((connection = tryGet()) != null) { return connection;}

        long startTime = System.nanoTime();
        long delay = timeout;

        while(isServiceStarted
                && freeConnectionsQueue.size() <= 0) {
            if(timeout <= 0) {
                wait();
            } else {
                delay = delay - System.nanoTime() + startTime;
                if(delay <= 0) {
                    break;
                }
                wait(delay / 1_000_000, (int)(delay % 1_000_000));
            }
        }

        return tryGet();
    }

    /**
     * blocking get()
     * @return {@code null} if service not started,
     * {@code T} otherwise
     * @throws InterruptedException
     */
    @Override
    public synchronized T get() throws InterruptedException {
        return mainGet(0);
    }

    /**
     * blocking get()
     * @return {@code null} if service not started or timeout exceed,
     * {@code T} otherwise
     * @throws InterruptedException
     */
    @Override
    public synchronized T get(long time, TimeUnit unit) throws InterruptedException {
        if(time <= 0) {
            return mainGet(0);
        }
        return mainGet(unit.toNanos(time));
    }

    /**
     * non-blocking get()
     * @return {@code null} if service not started or service has no free connections,
     * {@code T} otherwise
     */
    @Override
    public synchronized T tryGet() {
        if(!isServiceStarted) { return null;}
        T connection = freeConnectionsQueue.poll();
        if(connection != null) {
            return connection;
        }

        if(aliveConnections < maxCapacity) {
            aliveConnections++;
            return openNewConnection();
        }
        return null;
    }

    @Override
    public synchronized void release(T object) {
        if(object == null) { return;}
        if(!timeToKillConnections.keySet().contains(object)) { return;}
        if (freeConnectionsQueue.contains(object)) { return;}
        freeConnectionsQueue.addFirst(object);
        timeToKillConnections.put(object, System.nanoTime() + keepAliveTime);
        notifyAll();
    }

    @Override
    public synchronized void start() {
        if(connectionKiller == null) {
            connectionKiller = new ConnectionKiller();
            connectionKiller.setDaemon(true);
            connectionKiller.start();
        }
        connectionKiller.wakeUp();
        isServiceStarted = true;
    }

    /**
     * stop the service, all get() methods return {@code null}
     */
    @Override
    public synchronized void stop() {
        isServiceStarted = false;
        connectionKiller.goToSleep();
        notifyAll();
    }

    private class ConnectionKiller extends Thread {
        private AtomicBoolean isThreadStarted = new AtomicBoolean(false);

        @Override
        public void run() {
            while(!Thread.interrupted()) {
                try {
                    while((isThreadStarted.get()) || (getAliveConnections() > minCapacity)) {
                        long timeToSleep = 0;

                        synchronized(ConnectionPoolService.this) {
                            T eldestConnection = freeConnectionsQueue.peekLast();

                            if ((aliveConnections > minCapacity) && (eldestConnection != null)) {
                                Long timeToKill = timeToKillConnections.get(eldestConnection);
                                if (timeToKill != null) {
                                    long delay = timeToKill - System.nanoTime();
                                    if (delay > 0) {
                                        timeToSleep = delay;
                                    } else {
                                        tryRemoveConnection(eldestConnection);
                                    }
                                }
                            } else {
                                timeToSleep = keepAliveTime;
                            }
                        }
                        if(timeToSleep > 0) {
                            sleepNanos(timeToSleep);
                        }
                    }
                    synchronized(this) {
                        while(!isThreadStarted.get()) {
                            wait();
                        }
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        }

        private void tryRemoveConnection(T eldestConnection) {
            synchronized (ConnectionPoolService.this) {
                if (freeConnectionsQueue.remove(eldestConnection)) {
                    connectionSource.close(eldestConnection);
                    timeToKillConnections.remove(eldestConnection);
                    aliveConnections--;
                }
            }
        }

        private int getAliveConnections() {
            synchronized (ConnectionPoolService.this) {
                return aliveConnections;
            }
        }

        public synchronized void wakeUp() {
            isThreadStarted.set(true);
            notifyAll();
        }

        public synchronized void goToSleep() {
            isThreadStarted.set(false);
        }

        private void sleepNanos(long sleep) throws InterruptedException {
            Thread.sleep(sleep / 1_000_000, (int)(sleep % 1_000_000));
        }
    }
}